package cn.ywang.activemq;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * Pub/Sub 发布订阅
 * 持久订阅：订阅者能够获得未激活之前的消息
 *
 * @author: BeatificWang
 * @date: 2018/12/22 17:04
 */
public class PubSubPersistentProducer {
	public static void main(String[] args) {
		ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
		Connection connection = null;
		try {
			String clientID = "wang";
			// 创建连接
			connection = factory.createConnection();
			connection.setClientID(clientID);
			connection.start();
			// 第一个参数 transacted 代表是否开启事务 true为开启 false为关闭
			// 非事务活动 不需要使用session.commit()和session.rollback()
			// 第二个参数 acknowledgeMode 代表session的确认方式
			// AUTO_ACKNOWLEDGE 自动确认 不需要使用session.commit()
			// CLIENT_ACKNOWLEDGE 客户端确认 只适用于Consumer端
			// DUPS_OK_ACKNOWLEDGE 延迟确认 批量处理

			// SESSION_TRANSACTED session确认
			((ActiveMQConnectionFactory) factory).setUseAsyncSend(Boolean.TRUE);
			Session session = connection.createSession(Boolean.TRUE, Session.CLIENT_ACKNOWLEDGE);
			// 目的地
			// Pub/Sub
			Topic topic = session.createTopic("myTopic");
			// 消息生产者
			MessageProducer producer = session.createProducer(topic);
			// 消息
			Message msg = session.createTextMessage("hello, wang");
			// 持久化消息，防止消息丢失
			msg.setJMSDeliveryMode(DeliveryMode.PERSISTENT);
			producer.send(msg);
			// 消息发送
			session.commit();
			// 消息回滚
			//<p> session.rollback() <p>
			session.close();
		} catch (JMSException e) {
			e.printStackTrace();
		} finally {
			if (connection != null) {
				try {
					connection.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}

}
